package utils

import (
	"fmt"
	"os"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

type async_rec struct {
	cb          SubscribeCallback
	sender      interface{}
	channelid   string
	subscribeid interface{}
	data        interface{}
	tag         interface{}
	channel     *subscribeChannel
}

type subscribeChannel struct {
	fail_cnt        int32
	id              string
	listMap         sync.Map
	owner           *SubscribeCenter
	dispflag        int32
	lastdisp_t      time.Time
	lastdispstr     string
	lastdispwarn    string
	TaskPushFailMsg string
	popMonitor      *MonitorSpeedNTimeRec
	pushMonitor     *MonitorSpeedNTimeRec
}

/*
*
  - 主要功能: 发布订阅模式
  - ConfigASyncDispatch
    true: 使用GroupTask异步分发(执行)任务, 同一个订阅ID排队执行(allTopic包含在内 如果订阅id不同会异步执行)
    false: 使用调用线程依次执行任务阻塞执行
  - DispMode:
  - 0: 多线程投递可能会失败返回-1.
  - 1: 允许多线程投递. 不处理通道的lastdisp信息(因为多线程并发投递,处理了没有意义)
  - 2: 直接调用会掉, 直接进行分发 多线程不会返回-1  会掉会遭遇并发
  - 创建一个任务对象
*/
type SubscribeCenter struct {
	asynExec bool //  异步执行, 使用GroupTask同一个订阅ID排队执行(allTopic包含在内 如果订阅id不同会异步执行)

	DispMode byte
	task     *MultiGroupTask
	listMap  sync.Map

	AllTopicDispFlag int8
	allTopic         *subscribeChannel

	OnFilter SubscribeCallback

	Monitor_Exec   *MonitorSpeedNTimeRec // 执行次数(如果有多个订阅者会多次)
	Monitor_Push   *MonitorSpeedNTimeRec // 压入次数
	OnSubChannel   func(channelid string, n int)
	OnUnSubChannel func(channelid string, n int)

	/**
	  在执行分发的纤程中触发
	*/
	OnPushFail func(sender interface{}, channelid string, subscribeid interface{}, data interface{}, tag interface{})

	/**
	  任务推送的警告
	*/
	LastWarn_TaskPush string

	// 用于做信息输出描述
	Id   string
	Desc string
}

func newSubscribeChannel() *subscribeChannel {
	rval := &subscribeChannel{
		popMonitor:  NewMonitorSpeedNTimeRec(),
		pushMonitor: NewMonitorSpeedNTimeRec(),
	}
	return rval
}

/*
**

	注意不能多线程投递，多线程投递可能会失败
*/
func NewSubscribeCenter() *SubscribeCenter {
	rval := &SubscribeCenter{
		asynExec:         false,
		DispMode:         0,
		AllTopicDispFlag: 1,
		task:             DefaultWorkers(),
		allTopic:         newSubscribeChannel(),
		Monitor_Exec:     NewMonitorSpeedNTimeRec(),
		Monitor_Push:     NewMonitorSpeedNTimeRec(),
	}
	rval.allTopic.id = "allTopic"
	rval.allTopic.owner = rval
	return rval
}

func NewSubscribeCenterASync() *SubscribeCenter {
	rval := NewSubscribeCenter()
	rval.ConfigASyncDispatch(true)
	rval.DispMode = 1
	return rval
}

func _DateTimeString2(val time.Time) string {
	if val.IsZero() {
		return ""
	} else {
		return val.Format("2006-01-02 15:04:05")
	}
}

func (this *SubscribeCenter) CreateSingleTask() {
	this.task = NewMultiGroupTask()
}

func (this *SubscribeCenter) ConfigInnerTask(minN, maxN, maxQueueSize int32) {
	this.task.ConfigMinWork(minN)
	this.task.ConfigMaxWork(maxN)
	this.task.ConfigChannelMaxQueueSize(maxQueueSize)
	this.task.Start()
}

/**
 * true:  开启异步执行订阅者回调, 相同的id 排队进行分发, 如果是异步执行 回调的isBreak和返回值将会忽略
 * false: 调用者线程 同步阻塞执行订阅者回调, 默认值
 */
func (this *SubscribeCenter) ConfigASyncDispatch(val bool) *SubscribeCenter {
	this.asynExec = val
	if this.asynExec {
		this.task.Start()
	} else {
		if this.task == DefaultWorkers() { // 默认的不运行操作
			return this
		}
		this.task.Stop()
	}
	return this
}

func (this *SubscribeCenter) ListChannelEx(filter func(key, value interface{}) bool, less_sortfunc func(iItm, jItm interface{}) bool) []interface{} {
	rval := make([]interface{}, 0, this.ChannelCount())
	this.listMap.Range(func(key, value interface{}) bool {
		if filter != nil {
			if filter(key, value) {
				rval = append(rval, value)
			}
		} else {
			rval = append(rval, value)
		}

		return true
	})

	if less_sortfunc != nil {
		sort.Slice(rval, func(i, j int) bool {
			return less_sortfunc(rval[i], rval[j])
		})
	}

	return rval
}

func (this *SubscribeCenter) ConfigASyncTaskMaxWorkNum(n int32) *SubscribeCenter {
	if this.task != DefaultWorkers() {
		this.task.ConfigMaxWork(n)
	}
	return this
}

func (this *SubscribeCenter) ConfigASyncTaskDefaultMaxQueue(queue_size int32) *SubscribeCenter {
	if this.task != DefaultWorkers() {
		this.task.ConfigChannelMaxQueueSize(queue_size)
	}
	return this
}

func (this *SubscribeCenter) GetASyncTaskDefaultMaxQueue() int32 {
	return this.task.channel_max_queue_size
}

func (this *SubscribeCenter) ConfigDispMode(val byte) *SubscribeCenter {
	this.DispMode = val
	return this
}

/***
没有意义
*/
//func (this *SubscribeCenter) Remain() int64 {
//	return 0
//}

func (this *SubscribeCenter) onMultiTaskCb(sender *MultiGroupTaskRec) {
	if len(sender.Args) > 0 {
		rec := sender.Args[0].(*async_rec)
		isBreak := false
		m := rec.channel.popMonitor
		if m != nil {
			m.Delta(1)
		}
		this.Monitor_Exec.Delta(1)
		if this.OnFilter != nil {
			if this.OnFilter(rec.sender, rec.channelid, rec.subscribeid, rec.data, rec.tag, &isBreak) {
				rec.cb(rec.sender, rec.channelid, rec.subscribeid, rec.data, rec.tag, &isBreak)
			}
		} else {
			rec.cb(rec.sender, rec.channelid, rec.subscribeid, rec.data, rec.tag, &isBreak)
		}
	}
}

func (this *SubscribeCenter) doCallBack(channel *subscribeChannel, Cb SubscribeCallback, sender interface{}, channelid string, subscribeid interface{}, data interface{}, tag interface{}, isBreak *bool) bool {
	if this.asynExec {
		rec := &async_rec{
			cb:          Cb,
			sender:      sender,
			channelid:   channelid,
			subscribeid: subscribeid,
			data:        data,
			tag:         tag,
			channel:     channel,
		}
		err := this.task.PostTaskArgs(subscribeid, this.onMultiTaskCb, rec)
		if err != nil {
			atomic.AddInt32(&channel.fail_cnt, 1)

			onPushFail := this.OnPushFail
			if onPushFail != nil {
				onPushFail(sender, channelid, subscribeid, data, tag)
			} else {

				idstr := TrimUtf8Str(fmt.Sprintf("%v", subscribeid), 80)
				datastr := ""
				if buf, ok := data.([]byte); ok {
					datastr = BufToHexStr(buf, 128, "")
				} else {
					datastr = fmt.Sprintf("%v", data)
				}
				strMsg := fmt.Sprintf("[%s][%s]压入订阅者[%s](channelid:%s, data:%v), 处理任务队列时异常:%s\n", NowString(), this.Id, idstr, channelid, datastr, err.Error())
				channel.TaskPushFailMsg = strMsg
				this.LastWarn_TaskPush = strMsg

				//fmt.Fprintf(os.Stderr, "[%s][%s]压入订阅者[%s](channelid:%s, data:%v), 处理任务队列时异常:%s\n", NowString(), this.Id, idstr, channelid, datastr, err.Error())
			}

			return false
		} else {
			m := channel.pushMonitor
			if m != nil {
				m.Delta(1)
			}
			return true
		}
	} else {
		this.Monitor_Exec.Delta(1)
		if this.OnFilter != nil {
			if this.OnFilter(sender, channelid, subscribeid, data, tag, isBreak) {
				return Cb(sender, channelid, subscribeid, data, tag, isBreak)
			} else {
				return false
			}
		} else {
			return Cb(sender, channelid, subscribeid, data, tag, isBreak)
		}
	}
}

func (this *SubscribeCenter) checkGetChannel(channelid string, newIfNotfound bool) *subscribeChannel {
	if v, loaded := this.listMap.Load(channelid); loaded {
		return v.(*subscribeChannel)
	}

	if newIfNotfound {
		channel := newSubscribeChannel()
		channel.id = channelid
		channel.owner = this
		if v, loaded := this.listMap.LoadOrStore(channelid, channel); loaded {
			return v.(*subscribeChannel)
		}
		return channel
	} else {
		return nil
	}
}

/**
 * 如果是异步执行 回调的isBreak和返回值将会忽略, 异步执行保证同一个id将会按照顺序执行
 * 同步执行, 取决于Dispatch, 如果Disptach是纤程执行,可能会导致回调重入
 */
func (this *SubscribeCenter) SubscribeChannel(channelid string, id interface{}, cb SubscribeCallback) {
	channel := this.checkGetChannel(channelid, true)
	channel.subscribe(id, cb)
	if this.OnSubChannel != nil {
		this.OnSubChannel(channelid, channel.Count())
	}
}

/*
*
订阅所有的通道数据
*/
func (this *SubscribeCenter) SubscribeAllChannel(id interface{}, cb SubscribeCallback) {
	channel := this.allTopic
	channel.subscribe(id, cb)
	if this.OnSubChannel != nil {
		this.OnSubChannel("*", channel.Count())
	}
}

/*
*
取消订阅所有通道的数据
*/
func (this *SubscribeCenter) UnSubscribeAllChannel(id interface{}) bool {
	channel := this.allTopic
	r := channel.unsubscribe(id)
	if r && this.OnUnSubChannel != nil {
		this.OnUnSubChannel("*", channel.Count())
	}
	return r

}

func (this *SubscribeCenter) ChannelSubscribeCount(channelid string) int {
	channel := this.checkGetChannel(channelid, false)
	if channel == nil {
		return 0
	}
	return channel.Count()
}

func (this *SubscribeCenter) UnsubscribeChannelV0(channelid string, id interface{}) int {
	channel := this.checkGetChannel(channelid, false)
	if channel == nil {
		return -1
	}
	r := channel.unsubscribe(id)
	if r && this.OnUnSubChannel != nil {
		this.OnUnSubChannel(channelid, channel.Count())
	}
	return channel.Count()
}

func (this *SubscribeCenter) UnsubscribeChannel(channelid string, id interface{}) bool {
	channel := this.checkGetChannel(channelid, false)
	if channel == nil {
		return false
	}
	r := channel.unsubscribe(id)
	if r && this.OnUnSubChannel != nil {
		this.OnUnSubChannel(channelid, channel.Count())
	}
	return r
}

/*
**

	 如果多线程并发,过快,可能会返回-1
	0: 通道不存在, 没有被订阅

-1: 发送太快, 频繁, 或者阻塞.
*/
func (this *SubscribeCenter) DispatchData(channelid string, sender interface{}, data interface{}, tag interface{}) (rval int) {
	this.Monitor_Push.Delta(1)
	if this.AllTopicDispFlag == 1 {
		switch this.DispMode {
		case 1:
			rval = this.allTopic.dispatchDataMode1(sender, channelid, data, tag)
		case 2:
			rval = this.allTopic.dispatchDataMode2(sender, channelid, data, tag)
		default:
			rval = this.allTopic.dispatchDataMode0(sender, channelid, data, tag)
		}
	}

	channel := this.checkGetChannel(channelid, false)
	if channel == nil {
		return
	}
	switch this.DispMode {
	case 1:
		rval += channel.dispatchDataMode1(sender, channelid, data, tag)
	case 2:
		rval += channel.dispatchDataMode2(sender, channelid, data, tag)
	default:
		rval += channel.dispatchDataMode0(sender, channelid, data, tag)
	}

	return
}

func (this *SubscribeCenter) DispatchSimple(channelid string, data interface{}) int {
	return this.DispatchData(channelid, nil, data, nil)
}

func (this *SubscribeCenter) GetChannelIDListEx(timeout time.Duration) []string {
	rval := make([]string, 0)
	this.listMap.Range(func(key, value interface{}) bool {
		ch := value.(*subscribeChannel)
		if !ch.lastdisp_t.IsZero() && time.Since(ch.lastdisp_t) < timeout {
			rval = append(rval, ch.id)
		}
		return true
	})
	return rval
}

func (this *SubscribeCenter) GetChannelIDList() []string {
	rval := make([]string, 0)
	this.listMap.Range(func(key, value interface{}) bool {
		ch := value.(*subscribeChannel)
		rval = append(rval, ch.id)
		return true
	})
	return rval
}

func (this *SubscribeCenter) String() string {
	return fmt.Sprintf("SubscribeCenter-%p", this)
}

func (this *SubscribeCenter) InnerTask() *MultiGroupTask {
	return this.task
}

func (this *SubscribeCenter) TaskGroupStatusString(id string) string {
	return this.task.GroupStatus(id)
}

func (this *SubscribeCenter) ChannelStatusString(id string) string {
	if v, loaded := this.listMap.Load(id); loaded {
		ch := v.(*subscribeChannel)
		var sb strings.Builder
		sb.WriteString("id:\t")
		sb.WriteString(ch.id)
		sb.WriteString("\r\n")

		sb.WriteString(fmt.Sprintf("sub-n:\t%d\r\n", ch.Count()))

		sb.WriteString("push:\t")
		sb.WriteString(ch.pushMonitor.SpeedIO())
		sb.WriteString("\r\n")

		sb.WriteString("pop:\t")
		sb.WriteString(ch.popMonitor.SpeedIO())
		sb.WriteString("\r\n")

		sb.WriteString(fmt.Sprintf("fail:\t%d\r\n", ch.fail_cnt))
		sb.WriteString(fmt.Sprintf("lastdisp_t:\t%s\r\n", _DateTimeString2(ch.lastdisp_t)))
		sb.WriteString(fmt.Sprintf("lastdisp_str:\t%s\r\n", ch.lastdispstr))
		return sb.String()
	}
	return ""
}

func (this *SubscribeCenter) ChannelCount() int {
	r := 0
	this.listMap.Range(func(key, value interface{}) bool {
		r++
		return true
	})
	return r
}

func (this *SubscribeCenter) ChannelCountEx() (total, livecnt int) {
	total = 0
	livecnt = 0
	this.listMap.Range(func(key, value interface{}) bool {
		ch := value.(*subscribeChannel)
		if ch.Count() > 0 {
			livecnt++
		}
		total++
		return true
	})
	return
}

func (this *SubscribeCenter) ChannelDetails(max int) string {
	r := 0

	lst := this.ListChannelEx(nil, func(iItm, jItm interface{}) bool {
		chi := iItm.(*subscribeChannel)
		chj := jItm.(*subscribeChannel)
		if chi.Count() > 0 && chj.Count() > 0 { // 是否有订阅者
			if chi.fail_cnt > 1024 && chj.fail_cnt > 1024 {
				return chi.pushMonitor.CurrN >= chj.pushMonitor.CurrN
			} else {
				return chi.fail_cnt > chj.fail_cnt
			}
		} else {
			if chi.Count() <= 0 {
				return false
			}
			return true
		}
	})

	var sb strings.Builder

	for i := 0; i < len(lst); i++ {
		ch := lst[i].(*subscribeChannel)

		sb.WriteString("id:\t")
		sb.WriteString(ch.id)
		sb.WriteString("\r\n")

		sb.WriteString(fmt.Sprintf("sub-n:\t%d\r\n", ch.Count()))

		sb.WriteString("push:\t")
		sb.WriteString(ch.pushMonitor.SpeedIO())
		sb.WriteString("\r\n")

		sb.WriteString("pop:\t")
		sb.WriteString(ch.popMonitor.SpeedIO())
		sb.WriteString("\r\n")

		sb.WriteString(fmt.Sprintf("fail:\t%d\r\n", ch.fail_cnt))
		sb.WriteString(fmt.Sprintf("lastdisp_t:\t%s\r\n", _DateTimeString2(ch.lastdisp_t)))
		sb.WriteString(fmt.Sprintf("lastdisp_str:\t%s\r\n", ch.lastdispstr))
		sb.WriteString("\r\n")
		r++
		if r >= max {
			break
		}
	}
	return sb.String()
}

func (this *SubscribeCenter) StatusSimple() string {
	total, livecnt := this.ChannelCountEx()
	if this.task != DefaultWorkers() {
		return fmt.Sprintf("channel total/live(sub > 0):%d/%d, task-warn:%s, task-worker:%s", total, livecnt, this.LastWarn_TaskPush, this.task.StatusSimpleString())
	} else {
		return fmt.Sprintf("channel total/live(sub > 0):%d/%d", total, livecnt)
	}

}

func (this *SubscribeCenter) LatestChannel(t time.Duration) string {

	// dispMode : 0
	var sb strings.Builder
	sb.WriteString(fmt.Sprintf("channel num:%d\r\n", SyncMapLength(&this.listMap)))
	sb.WriteString("latestChannel:\r\n")
	this.listMap.Range(func(key, value interface{}) bool {
		channel := value.(*subscribeChannel)
		sub_n := SyncMapLength(&channel.listMap)
		if time.Since(channel.lastdisp_t) < t || sub_n > 0 {
			sb.WriteString(fmt.Sprintf("%s[%s], sub:%d\r\n", channel.id, DateTimeString2(channel.lastdisp_t),
				sub_n))
		}

		return true
	})

	return sb.String()
}

func (this *SubscribeCenter) StatusString() string {

	// dispMode : 0
	var sb strings.Builder
	if this.DispMode == 0 {
		sb.WriteString(fmt.Sprintf("channel num:%d\r\n", SyncMapLength(&this.listMap)))
		sb.WriteString("阻塞的分发通道:\r\n")
		this.listMap.Range(func(key, value interface{}) bool {
			channel := value.(*subscribeChannel)

			if channel.dispflag == 1 {
				sb.WriteString(channel.lastdispstr)
				sb.WriteString("\r\n")
			}
			return true
		})
	} else {
		sb.WriteString("task status:\r\n")
		sb.WriteString(this.task.StatusString())
	}

	return sb.String()
}

// ---------- Channel

func (this *subscribeChannel) Count() int {
	return SyncMapLength(&this.listMap)
}

func (this *subscribeChannel) subscribe(id interface{}, cb SubscribeCallback) {
	this.listMap.Store(id, cb)
}

func (this *subscribeChannel) unsubscribe(id interface{}) bool {
	if _, loaded1 := this.listMap.Load(id); loaded1 {
		this.listMap.Delete(id)
		return true
	} else {
		return false
	}
}

/*
*
允许多线程投递
不处理 lastdisp信息
*/
func (this *subscribeChannel) dispatchDataMode1(sender interface{}, channelid string, data interface{}, tag interface{}) int {
	isBreak := false
	r := 0

	this.lastdisp_t = time.Now()

	// 订阅者列表
	this.listMap.Range(func(key, value interface{}) bool {
		cb := value.(SubscribeCallback)
		if cb == nil {
			return true
		}
		if this.owner.doCallBack(this, cb, sender, channelid, key, data, tag, &isBreak) {
			r++
		}

		return !isBreak
	})
	return r

}

/*
*

	直接分发, 会掉会多线程进入
*/
func (this *subscribeChannel) dispatchDataMode2(sender interface{}, channelid string, data interface{}, tag interface{}) int {
	atomic.StoreInt32(&this.dispflag, 1)
	defer func() { atomic.StoreInt32(&this.dispflag, 0) }()
	this.lastdisp_t = time.Now()
	isBreak := false
	r := 0

	// 订阅者列表
	this.listMap.Range(func(key, value interface{}) bool {
		cb := value.(SubscribeCallback)
		if cb == nil {
			return true
		}
		if this.owner.doCallBack(this, cb, sender, channelid, key, data, tag, &isBreak) {
			r++
		}
		return !isBreak
	})
	return r

}

/**
 * 一个通道分发时,只能同时一个进入, 避免阻塞
 *   如果阻塞了  返回-1
 */
func (this *subscribeChannel) dispatchDataMode0(sender interface{}, channelid string, data interface{}, tag interface{}) int {
	if atomic.CompareAndSwapInt32(&this.dispflag, 0, 1) {
		defer func() { this.dispflag = 0 }()
		this.lastdisp_t = time.Now()
		isBreak := false
		r := 0
		//this.lastdispstr = fmt.Sprintf("[%s] goid:%d, channelid:%s\r\n", NowString(), GetCurrentGoRoutineID(),
		//	CutIfMore(this.id, 10, "..."))

		// 订阅者列表
		this.listMap.Range(func(key, value interface{}) bool {

			//k := CutIfMore(fmt.Sprintf("%v", key), 10, "...")
			cb := value.(SubscribeCallback)
			if cb == nil {
				return true
			}
			//this.lastdispstr += fmt.Sprintf("  ->[%s]key:%s\r\n", NowString(), k)
			if this.owner.doCallBack(this, cb, sender, channelid, key, data, tag, &isBreak) {
				r++
			}
			return !isBreak
		})
		return r
	} else {
		if time.Now().Sub(this.lastdisp_t).Seconds() > 10 {
			str := fmt.Sprintf("[%s][警告]通道[%s]逻辑执行堵塞, lastdispinfo:%s\r\n", NowString(), this.id, this.lastdispstr)
			this.lastdispwarn = str
			fmt.Fprintf(os.Stderr, "[SubscribeCenter]%s", str)
		}
		return -1
	}

}
